package com.weimin.mqtt.server.handler;

import com.weimin.mqtt.server.management.MqttSessionManager;
import com.weimin.mqtt.server.model.MqttServerConfig;
import com.weimin.mqtt.server.model.MqttSessionPublishMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Slf4j
public class MqttChannelInboundHandler extends ChannelInboundHandlerAdapter {
    // 客户端连接对象
    private final Map<String, Channel> connectMap = new ConcurrentHashMap<>();

    private final MqttServerConfig mqttServerConfig;
    private final MqttSessionManager mqttSessionManager;

    public MqttChannelInboundHandler(MqttServerConfig config, MqttSessionManager mqttSessionManager) {
        this.mqttServerConfig = config;
        this.mqttSessionManager = mqttSessionManager;
    }


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg == null) {
            return;
        }
        try {
            MqttMessage mqttMessage = (MqttMessage) msg;
            MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
            if (mqttFixedHeader == null) {
                log.info("接收到MQTT消息，但是header为null，mqttMessage={}", mqttMessage);
                return;
            }
            Channel channel = ctx.channel();
            String connectIp = getSourceIp(ctx);
            switch (mqttFixedHeader.messageType()) {
                case CONNECT:
                    //	这里直接返回一个CONNACK消息在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
                    String clientId = ((MqttConnectPayload) mqttMessage.payload()).clientIdentifier();
                    String userName = ((MqttConnectPayload) mqttMessage.payload()).userName();
                    byte[] password = ((MqttConnectPayload) mqttMessage.payload()).passwordInBytes();
                    log.info("MQTT 用户连接,clientId---{},username---{},ip---{}", clientId, userName, connectIp);
                    IMqttUserAuthHandler userAuthHandler = mqttServerConfig.getUserAuthHandler();
                    if (userAuthHandler != null) {
                        if (!userAuthHandler.verify(clientId, userName, password)) {
                            log.info("MQTT 用户名密码验证错误,客户端ID:{} 用户名: {} 密码: {}", clientId, userName, new String(password, StandardCharsets.UTF_8));
                            ctx.close();
                            break;
                        }
                    }
                    connack(channel, mqttMessage);
                    mqttSessionManager.online(connectIp, channel.id().asLongText(), this::publish, clientId, userName, password);
                    break;
                case PUBLISH:
                    //客户端发布消息
                    String topic = ((MqttPublishVariableHeader) mqttMessage.variableHeader()).topicName();
                    if (mqttMessage.payload() == null) {
                        log.error("收到异常数据,数据为空,主题为 {}", topic);
                        break;
                    }
                    byte[] data = ByteBufUtil.getBytes((ByteBuf) mqttMessage.payload());
                    mqttSessionManager.onMessage(new MqttSessionPublishMessage(connectIp, channel.id().asLongText(), topic, data, this::publish));
                    puback(channel, mqttMessage);
                    break;
                case PUBREL:
                    //	发布释放,PUBREL报文是对PUBREC报文的响应
                    pubcomp(channel, mqttMessage);
                    break;
                case SUBSCRIBE:
                    //	客户端订阅主题,客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅，每个订阅注册客户端关心的一个或多个主题。为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端
                    suback(channel, mqttMessage);
                    break;
                case UNSUBSCRIBE:
                    //	客户端取消订阅,客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题
                    unsuback(channel, mqttMessage);
                    break;
                case PINGREQ:
                    //	客户端发起心跳,客户端发送PINGREQ报文给服务端的,在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着，请求服务端发送 响应确认它还活着，使用网络以确认网络连接没有断开
                    pingresp(channel, mqttMessage);
                    break;
                case DISCONNECT:
                    //	客户端主动断开连接
                    ctx.close();
                    break;
                default:
                    break;
            }
        } catch (Exception e) {
            log.error("MQTT 通道数据处理异常", e);
            ctx.close();
        } finally {
            // 资源释放
            ReferenceCountUtil.release(msg);
        }
    }

    /**
     * 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("MQTT 服务通道异常,连接ID: {} IP: {}", ctx.channel().id().asLongText(), getSourceIp(ctx), cause);
        ctx.close();
    }

    /**
     * 客户端与服务端第一次建立连接时执行
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        String connectId = ctx.channel().id().asLongText();
        log.info("MQTT 通道连接 连接ID: {} IP: {}", connectId, getSourceIp(ctx));
        connectMap.put(connectId, ctx.channel());
    }

    /**
     * 客户端与服务端 断连时执行
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        ctx.close();
        String connectId = ctx.channel().id().asLongText();
        log.info("MQTT 通道断开 连接ID: {} IP: {}", connectId, getSourceIp(ctx));
        connectMap.remove(connectId);
        mqttSessionManager.offline(connectId);
    }

    /**
     * 服务端 当读超时时 会调用这个方法，SSL的时候也会触发SslCloseCompletionEvent，SslHandshakeCompletionEvent
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
        log.info("MQTT 通道读取超时 ID: {} IP: {} {}", ctx.channel().id().asLongText(), getSourceIp(ctx), evt.toString());
        if (evt instanceof IdleStateEvent) {
            ctx.close();
        }
    }


    /**
     * 发送消息给客户端，需传递通道连接ID，主题和内容
     */
    public boolean publish(MqttSessionPublishMessage mqttPublishMessage) {
        if (mqttPublishMessage == null || mqttPublishMessage.isEmpty()) {
            log.error("MQTT 通道异常消息 {}", mqttPublishMessage);
            return false;
        }
        String connectId = mqttPublishMessage.getConnectId();
        if (!connectMap.containsKey(connectId)) {
            log.error("MQTT 未找到对应连接通道,发送消息错误,连接ID: {},主题: {}", connectId, mqttPublishMessage.getTopic());
            return false;
        }
        Channel channel = connectMap.get(connectId);
        MqttPublishVariableHeader header = new MqttPublishVariableHeader(mqttPublishMessage.getTopic(), -1);
        ByteBuf payload = channel.alloc().buffer();
        byte[] bytes = mqttPublishMessage.getContent();
        payload.writeBytes(bytes);
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE,
                false, bytes.length + 5);
        channel.writeAndFlush(new MqttPublishMessage(mqttFixedHeader, header, payload));
        if (mqttPublishMessage.getContentType() == 1) {
            log.info("MQTT 发送消息 连接ID: {} 主题: {} 内容: {}", mqttPublishMessage.getConnectId(), mqttPublishMessage.getTopic(), new String(mqttPublishMessage.getContent()));
        } else {
            log.info("MQTT 发送消息 连接ID: {} 主题: {} 内容长度: {}", mqttPublishMessage.getConnectId(), mqttPublishMessage.getTopic(), mqttPublishMessage.getContent().length);
        }
        return true;
    }

    //------------------------------分割-----------------------------------------------------------------

    /**
     * 确认连接请求
     */
    private void connack(Channel channel, MqttMessage mqttMessage) {
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
        //	构建返回报文， 可变报头
        MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
        //	构建CONNACK消息体
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
        //logger.info("connack--"+connAck.toString());
        channel.writeAndFlush(connAck);
    }

    /**
     * 发布完成 qos2
     */
    private void pubcomp(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
        //logger.info("pubcomp--" + mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    /**
     * 根据qos发布确认
     */
    private void puback(Channel channel, MqttMessage mqttMessage) {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
        MqttQoS qos = mqttFixedHeaderInfo.qosLevel();
        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);

        switch (qos) {
            //	至多一次
            case AT_MOST_ONCE:
                break;
            //	至少一次
            case AT_LEAST_ONCE:
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                //	构建PUBACK消息体
                MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                //logger.info("AT_LEAST_ONCE puback--" + pubAck.toString());
                channel.writeAndFlush(pubAck);
                break;
            //	刚好一次
            case EXACTLY_ONCE:
                //	构建返回报文， 固定报头
                MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
                //	构建返回报文， 可变报头
                MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
                //logger.info("EXACTLY_ONCE puback--" + mqttMessageBack.toString());
                channel.writeAndFlush(mqttMessageBack);
                break;
            default:
                break;
        }
    }

    /**
     * 订阅确认
     */
    private void suback(Channel channel, MqttMessage mqttMessage) {
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(MqttTopicSubscription::topicName).collect(Collectors.toSet());
        mqttSessionManager.subTopics(channel.id().asLongText(), topics);
        List<Integer> grantedQosLevels = new ArrayList<>(topics.size());
        for (int i = 0; i < topics.size(); i++) {
            grantedQosLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        //	构建返回报文	有效负载
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQosLevels);
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2 + topics.size());
        //	构建返回报文	订阅确认
        MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
        //logger.info("suback--"+subAck.toString());
        channel.writeAndFlush(subAck);
    }

    /**
     * 取消订阅确认
     */
    private void unsuback(Channel channel, MqttMessage mqttMessage) {
        MqttUnsubscribeMessage mqttUnsubscribeMessage = (MqttUnsubscribeMessage) mqttMessage;
        mqttSessionManager.unSubTopics(channel.id().asLongText(), mqttUnsubscribeMessage.payload().topics());
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文	可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
        //	构建返回报文	取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
        //logger.info("unsuback--" + unSubAck.toString());
        channel.writeAndFlush(unSubAck);
    }

    /**
     * 心跳响应
     */
    private void pingresp(Channel channel, MqttMessage mqttMessage) {
        //	心跳响应报文	11010000 00000000  固定报文
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        channel.writeAndFlush(mqttMessageBack);
    }

    private static String getSourceIp(ChannelHandlerContext ctx) {
        InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
        return address.getAddress().getHostAddress();
    }
}
